package com.itzdm.ai.service.impl;

import com.alibaba.dashscope.aigc.generation.Generation;
import com.alibaba.dashscope.aigc.generation.GenerationOutput;
import com.alibaba.dashscope.aigc.generation.GenerationParam;
import com.alibaba.dashscope.aigc.generation.GenerationResult;
import com.alibaba.dashscope.common.Message;
import com.alibaba.dashscope.common.Role;

import com.alibaba.dashscope.exception.InputRequiredException;
import com.alibaba.dashscope.exception.NoApiKeyException;
import com.itzdm.ai.model.ChatContext;
import com.itzdm.ai.model.ChatMessage;
import com.itzdm.ai.service.ChatContextService;
import com.itzdm.ai.service.DashScopeService;
import com.itzdm.common.exception.BadRequestException;
import io.reactivex.Flowable;
import io.reactivex.disposables.Disposable;
import lombok.extern.slf4j.Slf4j;
import org.apache.catalina.connector.ClientAbortException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

/**
 * DashScope服务实现类
 */
@Slf4j
@Service
public class DashScopeServiceImpl implements DashScopeService {

    @Value("${dashscope.api-key}")
    private String apiKey;

    @Autowired
    private ChatContextService chatContextService;


    /**
     * 实现异步流式输出
     *
     * @param userId  用户ID
     * @param content 用户消息内容
     * @param emitter SSE发射器
     */
    @Override
    public void streamChat(Long userId, String content, SseEmitter emitter) {
        // 使用 AtomicBoolean 来跟踪连接状态
        AtomicBoolean isCompleted = new AtomicBoolean(false);
        // 使用 AtomicReference 存储 Disposable 对象，让它可以在 lambda 中使用
        AtomicReference<Disposable> disposableRef = new AtomicReference<>();

        try {
            // 首先需要将用户的信息添加到上下文中，并且获取用户上下文，准备向远程发送ai调用
            ChatContext chatContext = chatContextService.addUserMessage(userId, content);
            // 获取用户上下文,list最后的就是最新的消息
            List<ChatMessage> messageHistory = chatContext.getMessageHistory();
            // 由于我们需要将用户的消息和助手的消息都传递给DashScope，所以我们需要过滤掉推理内容
            List<ChatMessage> filterMsg = messageHistory.stream().filter(message -> {
                String role = message.getRole().toUpperCase();
                // 过滤掉推理内容
                return "USER".equals(role) || "SYSTEM".equals(role) ||
                        ("ASSISTANT".equals(role) && message.getReasoningContent() == null);
            }).collect(Collectors.toList());

            // 转换消息格式并调用DashScope API
            List<Message> dashScopeMessages = convertToDashScopeMessages(filterMsg);
            Generation generation = new Generation();
            // 构建GenerationParam对象
            GenerationParam param = GenerationParam.builder()
                    .apiKey(apiKey)
                    .model("deepseek-r1-distill-llama-70b")
                    .messages(dashScopeMessages)
                    .resultFormat(GenerationParam.ResultFormat.MESSAGE)
                    .incrementalOutput(true)
                    .build();

            // 流式调用
            StringBuilder responseContent = new StringBuilder();
            StringBuilder reasoningContent = new StringBuilder();

            // 添加连接状态检查
            emitter.onTimeout(() -> {
                log.info("SSE连接超时");
                isCompleted.set(true);
                safeCompleteEmitter(emitter);
                safeDisposeSubscription(disposableRef.get());
            });

            emitter.onError((ex) -> {
                log.info("SSE连接发生错误: {}", ex.getMessage());
                isCompleted.set(true);
                safeCompleteEmitter(emitter);
                safeDisposeSubscription(disposableRef.get());
            });

            emitter.onCompletion(() -> {
                log.info("SSE连接已完成");
                isCompleted.set(true);
                safeDisposeSubscription(disposableRef.get());
            });

            Flowable<GenerationResult> generationResultFlowable = generation.streamCall(param);
            Disposable disposable = generationResultFlowable.subscribe(
                    result -> {
                        try {
                            if (!isCompleted.get()) {
                                handleStreamResult(result, emitter, responseContent, reasoningContent, isCompleted);
                            }
                        } catch (Exception e) {
                            if (isClientAbortException(e)) {
                                log.info("客户端连接已断开，停止发送数据");
                                isCompleted.set(true);
                                safeDisposeSubscription(disposableRef.get());
                            } else {
                                log.error("处理流式结果时出错", e);
                                if (!isCompleted.get()) {
                                    try {
                                        emitter.send(SseEmitter.event()
                                                .name("error")
                                                .data("处理响应时出错"));
                                    } catch (IOException ignored) {
                                        // 忽略进一步的IO异常
                                    } finally {
                                        isCompleted.set(true);
                                        safeCompleteEmitter(emitter);
                                        safeDisposeSubscription(disposableRef.get());
                                    }
                                }
                            }
                        }
                    },
                    error -> {
                        if (isClientAbortException(error)) {
                            log.info("客户端连接已断开，停止处理");
                        } else {
                            log.error("流式调用出错", error);
                            if (!isCompleted.get()) {
                                try {
                                    emitter.send(SseEmitter.event()
                                            .name("error")
                                            .data("调用模型API出错"));
                                } catch (IOException ignored) {
                                    // 忽略进一步的IO异常
                                } finally {
                                    isCompleted.set(true);
                                    safeCompleteEmitter(emitter);
                                }
                            }
                        }
                    },
                    () -> {
                        if (!isCompleted.get()) {
                            try {
                                completeStream(userId, responseContent.toString(), reasoningContent.toString(), emitter, isCompleted);
                            } catch (Exception e) {
                                log.error("完成流式调用时出错", e);
                                safeCompleteEmitter(emitter);
                            }
                        }
                    }
            );

            // 将 disposable 存储到 AtomicReference 中
            disposableRef.set(disposable);

        } catch (Exception e) {
            log.error("调用dashscope API失败", e);
            if (!isCompleted.get()) {
                try {
                    emitter.send(SseEmitter.event().data("很抱歉，服务暂时不可用，请稍后再试。"));
                } catch (IOException ignored) {
                    // 忽略进一步的IO异常
                } finally {
                    isCompleted.set(true);
                    safeCompleteEmitter(emitter);
                }
            }
        }
    }

    /**
     * 检查异常是否为客户端中断异常
     */
    private boolean isClientAbortException(Throwable t) {
        Throwable cause = t;
        while (cause != null) {
            if (cause instanceof ClientAbortException ||
                    (cause instanceof IOException &&
                            cause.getMessage() != null &&
                            cause.getMessage().contains("你的主机中的软件中止了一个已建立的连接"))) {
                return true;
            }
            cause = cause.getCause();
        }
        return false;
    }

    /**
     * 安全地完成SSE发射器
     */
    private void safeCompleteEmitter(SseEmitter emitter) {
        try {
            emitter.complete();
        } catch (Exception e) {
            log.debug("完成SSE连接时发生异常", e);
        }
    }

    /**
     * 安全地取消订阅
     */
    private void safeDisposeSubscription(Disposable disposable) {
        if (disposable != null && !disposable.isDisposed()) {
            disposable.dispose();
        }
    }

    private void handleStreamResult(GenerationResult result, SseEmitter emitter,
                                    StringBuilder responseContent, StringBuilder reasoningContent,
                                    AtomicBoolean isCompleted) throws IOException {
        if (isCompleted.get()) {
            return;
        }

        GenerationOutput.Choice choice = result.getOutput().getChoices().get(0);
        String reasoning = choice.getMessage().getReasoningContent();
        String content = choice.getMessage().getContent();
        long timestamp = System.currentTimeMillis();

        try {
            if (reasoning != null && !reasoning.isEmpty()) {
                reasoningContent.append(reasoning);
                String encodedReasoning = reasoning.replace("\n", "\\n")
                        .replace("\r", "\\r");
                emitter.send(SseEmitter.event().data(encodedReasoning).name("reasoning").id(String.valueOf(timestamp)));
            }

            if (content != null && !content.isEmpty()) {
                responseContent.append(content);
                emitter.send(SseEmitter.event().data(content).name("content").id(String.valueOf(timestamp)));
            }
        } catch (IOException e) {
            if (isClientAbortException(e)) {
                log.info("客户端连接已断开，停止发送数据");
            } else {
                log.error("发送SSE事件失败", e);
            }
            isCompleted.set(true);
            throw e;
        }
    }

    private void completeStream(Long userId, String finalContent, String reasoningContent,
                                SseEmitter emitter, AtomicBoolean isCompleted) {
        if (isCompleted.get()) {
            return;
        }

        try {
            // 将最终回复和推理内容添加到上下文
            chatContextService.addAssistantMessageWithReasoning(userId, finalContent, reasoningContent);

            // 发送完成事件
            emitter.send(SseEmitter.event()
                    .name("complete")
                    .data("流式响应完成"));
        } catch (IOException e) {
            if (isClientAbortException(e)) {
                log.info("客户端连接已断开，无法发送完成事件");
            } else {
                log.error("发送完成事件失败", e);
            }
        } finally {
            isCompleted.set(true);
            safeCompleteEmitter(emitter);
        }
    }

    @Override
    public List<Message> convertToDashScopeMessages(List<ChatMessage> messages) {
        if (messages == null || messages.isEmpty()) {
            return new ArrayList<>();
        }
        // 不为空的话构建dashscope的消息类型
        return messages.stream().map(msg -> {
            Role role;
            if ("USER".equalsIgnoreCase(msg.getRole())) {
                role = Role.USER;
            } else if ("SYSTEM".equalsIgnoreCase(msg.getRole())) {
                role = Role.SYSTEM;
            } else {
                role = Role.ASSISTANT;
            }

            // 构建消息，不包含推理内容
            return Message.builder()
                    .role(role.getValue())
                    .content(msg.getContent())
                    // 不传递reasoningContent
                    .build();
        }).collect(Collectors.toList());
    }
}